package com.gzyj.flink.gps.map.beidoualarm;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.dto.alarm.Kafka_VehicleNormalAlarm;
import com.gzyj.flink.dto.alarm.MonitorClientAlarm;
import com.gzyj.flink.gps.common.AddressService;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.mapper.mysql.MonYanzhongweiguiSetMapper;
import com.gzyj.flink.gps.utils.TimeUtils;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummary;
import com.gzyj.flink.pojo.gps.BaobiaoAlarmsummaryRecord;
import com.gzyj.flink.pojo.mysql.set.MonYanzhongweiguiSet;
import com.gzyj.flink.veh.VehicleIncrement;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Set;

public class BeidouPlatformAlarmProcess extends RichFlatMapFunction<Kafka_VehicleNormalAlarm, Object> {


    private final static boolean zdplrpt = false;

    private AddressService addressService;

    SqlSessionFactory factory;
    LettuceConnectionFactory redisFactory;

    @Override
    public void open(Configuration parameters) throws Exception {

        addressService = new AddressService();
        factory = MySQLMybatisSessionFactory.getSqlSessionFactory();
        redisFactory = RedisSessionFactory.getRedisFactory();

    }

    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisSessionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
        redisTemplate.afterPropertiesSet();
        return redisTemplate;
    }

    public VehicleIncrement getVehicleInfo(String vehId, RedisTemplate<String, Object> clusterRedisTemplate) {

        Set<String> keys = clusterRedisTemplate.keys(RedisKeyPrefixConfig.RedisKey_VehicleInfo + "*:" + vehId + "*");
        VehicleIncrement vehicleInfo = null;
        if (CollectionUtil.isNotEmpty(keys)) {
            Object obj = clusterRedisTemplate.opsForValue().get(new ArrayList<String>(keys).get(0));
            if (obj == null) {
                return null;
            } else {
                vehicleInfo = JSONObject.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), VehicleIncrement.class);
            }
        } else {
            return null;
        }
        return vehicleInfo;
    }

    @Override
    public void flatMap(Kafka_VehicleNormalAlarm VehAlarm, Collector<Object> collector) throws Exception {
        // 平台报警
        double KeepTime = 0;
        int ChaoSuBi = 0;
        int BaoJingType = 0;
        String AlarmNote = "";

        List<Object> list = new ArrayList<>();
        SqlSession session = factory.openSession(true);
        try {
            MonYanzhongweiguiSetMapper monYanzhongweiguiSetMapper = session.getMapper(MonYanzhongweiguiSetMapper.class);

            MonYanzhongweiguiSet AlarmDeploy = monYanzhongweiguiSetMapper.selectByVehId(VehAlarm.getVehicleId());
            session.commit();
            RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
            VehicleIncrement vehinfo = getVehicleInfo(VehAlarm.getVehicleId(), clusterRedisTemplate);


            switch (VehAlarm.getNote()) {
                case "疲劳驾驶":
                    AlarmNote = "疲劳驾驶报警";
                    break;
                case "疲劳驾驶预警":
                    AlarmNote = "疲劳驾驶预警";
                    break;
                case "夜间禁行报警":
                    AlarmNote = "夜间行驶报警";
                    break;
                case "无数据报警":
                    AlarmNote = "无数据报警";
                    break;
                case "离线报警":
                    AlarmNote = "24小时不在线报警";
                    break;
                case "24小时不在线报警":
                    AlarmNote = "24小时不在线报警";
                    break;
                case "不定位报警":
                    AlarmNote = "不定位报警";
                    break;
                case "24小时不定位报警":
//                            AlarmNote = "24小时不在线报警";
                    break;
                case "高速禁行":
                    AlarmNote = "高速禁行报警";
                    break;
                case "异常掉线报警":
                    AlarmNote = "异常掉线报警";
                    break;
                case "累计疲劳驾驶报警":
                    AlarmNote="累计疲劳驾驶报警";
                    break;
            }


            switch (AlarmNote) {
                case "无数据报警":
                case "不定位报警":
                case "疲劳驾驶报警":
                case "夜间行驶报警":
                case "累计疲劳驾驶报警":
                    if (StringUtils.isEmpty(VehAlarm.getRoadName())) {
                        String road = addressService.getAddress(VehAlarm.getLatitude(), VehAlarm.getLongitude());
                        if (StringUtils.isNotEmpty(road)) {
                            VehAlarm.setRoadName(road);
                        }
                    }
                    break;
            }

            int analyzeMode = 1;
            if (zdplrpt && AlarmNote.equalsIgnoreCase("疲劳驾驶报警")) analyzeMode = 2;
            if (VehAlarm.getIsAlarm()) {
                if (AlarmNote.equalsIgnoreCase("疲劳驾驶报警")) {

                    //推送到kafka
                    MonitorClientAlarm vi = new MonitorClientAlarm();
                    vi.VehNo = vehinfo.getCheliangpaizhao();
                    vi.VehId = VehAlarm.getVehicleId();
                    vi.PlateColor = vehinfo.getChepaiyanse();
                    vi.AlarmFlag = 1;
                    vi.Time = VehAlarm.getTime();
                    vi.Note = AlarmNote;
                    vi.Longitude = VehAlarm.getLongitude();
                    vi.Latitude = VehAlarm.getLatitude();
                    vi.Velocity = VehAlarm.getVelocity();
                    vi.Angle = VehAlarm.getAngle();
                    vi.Local = VehAlarm.getLocal() ? 1 : 0;
                    vi.Limited = 0;
                    vi.RoadLimited = 0;
                    vi.RoadName = VehAlarm.getRoadName();
                    vi.AlarmGuid = VehAlarm.getAlarmGuid();
                    vi.Driver = VehAlarm.getDriver();
                    vi.alarmSource = 1;
                    vi.AlarmLevel = VehAlarm.getAlarmLevel();
                    list.add(vi);


                }

                BaobiaoAlarmsummary baobiaoAlarmsummary = new BaobiaoAlarmsummary();
                baobiaoAlarmsummary.setVehid(VehAlarm.getVehicleId());
                baobiaoAlarmsummary.setDeptId(VehAlarm.getDeptId());
                baobiaoAlarmsummary.setBegintime(TimeUtils.toDate(VehAlarm.getAlarmBeginTime()));
                baobiaoAlarmsummary.setDriverName(VehAlarm.getDriver());
                baobiaoAlarmsummary.setDriverNo(VehAlarm.getDriverNo());
                baobiaoAlarmsummary.setMaxspeed(VehAlarm.getVelocity());
                baobiaoAlarmsummary.setAlarmtype(AlarmNote);
                baobiaoAlarmsummary.setLongitude(BigDecimal.valueOf(VehAlarm.getLongitude()));
                baobiaoAlarmsummary.setLatitude(BigDecimal.valueOf(VehAlarm.getLatitude()));
                baobiaoAlarmsummary.setVelocity(VehAlarm.getVelocity());
                baobiaoAlarmsummary.setAngle(VehAlarm.getAngle());
                baobiaoAlarmsummary.setLocal(VehAlarm.getLocal() ? 1 : 0);
                baobiaoAlarmsummary.setElevation(BigDecimal.valueOf(VehAlarm.getElevation()));
                baobiaoAlarmsummary.setLimited(0);
                baobiaoAlarmsummary.setTime(TimeUtils.toDate(VehAlarm.getAlarmBeginTime()));
                baobiaoAlarmsummary.setRoadName(VehAlarm.getRoadName());
                baobiaoAlarmsummary.setRoadLevel("");
                baobiaoAlarmsummary.setRoadLimited(0);
                baobiaoAlarmsummary.setDisposealarmname("");
                baobiaoAlarmsummary.setDisposealarmtime(TimeUtils.toDate(VehAlarm.getTime()));
                baobiaoAlarmsummary.setAlarmcl("短信发送");
                baobiaoAlarmsummary.setAnalyzemode(analyzeMode);
                baobiaoAlarmsummary.setStatus(0);
                baobiaoAlarmsummary.setSyntime(new Date());
                baobiaoAlarmsummary.setPlatenumber(vehinfo.getCheliangpaizhao());
                baobiaoAlarmsummary.setColor(vehinfo.getChepaiyanse());
                baobiaoAlarmsummary.setCompany(vehinfo.getDeptName());
                baobiaoAlarmsummary.setOperattype(vehinfo.getShiyongxingzhi());
                baobiaoAlarmsummary.setCalctime(VehAlarm.getCalcTime() != null ? TimeUtils.toDate(VehAlarm.getCalcTime()) : null);
                baobiaoAlarmsummary.setAlarmguid(VehAlarm.getAlarmGuid());

                list.add(baobiaoAlarmsummary);

                if (!AlarmNote.equalsIgnoreCase("24小时不在线报警")) {
                    BaobiaoAlarmsummaryRecord baobiaoAlarmsummaryRecord = new BaobiaoAlarmsummaryRecord();

                    baobiaoAlarmsummaryRecord.setVehid(VehAlarm.getVehicleId());
                    baobiaoAlarmsummaryRecord.setBegintime(TimeUtils.toDate(VehAlarm.getTime()));
                    baobiaoAlarmsummaryRecord.setAlarmtype(AlarmNote);
                    baobiaoAlarmsummaryRecord.setLongitude(BigDecimal.valueOf(VehAlarm.getLongitude()));
                    baobiaoAlarmsummaryRecord.setLatitude(BigDecimal.valueOf(VehAlarm.getLatitude()));
                    baobiaoAlarmsummaryRecord.setVelocity(VehAlarm.getVelocity());
                    baobiaoAlarmsummaryRecord.setAngle(VehAlarm.getAngle());
                    baobiaoAlarmsummaryRecord.setRoadLevel("0");
                    baobiaoAlarmsummaryRecord.setRoadLimited(0);
                    baobiaoAlarmsummaryRecord.setKeeptime(0);
                    baobiaoAlarmsummaryRecord.setLevel(1);
                    baobiaoAlarmsummaryRecord.setAlarmreportid(VehAlarm.getAlarmGuid());
                    list.add(baobiaoAlarmsummaryRecord);

                }

            } else {
                KeepTime = TimeUtils.getMinusSecond(VehAlarm.getAlarmEndTime(), VehAlarm.getAlarmBeginTime());
                if (KeepTime < 0) KeepTime = 0;
                // 获取BaoJingType 安标使用,作用不明 不定位报警,无数据报警=0,其他报警=1
                switch (VehAlarm.getNote()) {
                    case "不定位报警":
                    case "无数据报警":
                        BaoJingType = 0;
                        break;
                    default:
                        BaoJingType = 1;
                        break;
                }
                //严重违规计算
                int status = 2;
                switch (AlarmNote) {
                    case "疲劳驾驶报警": {
                        int plsj = -1;
                        int plsj1 = -1;
                        int plsj2 = -1;
                        if (AlarmDeploy != null) {
                            status = 1;
                            plsj = AlarmDeploy.getPlcxsj() == null ? -1 : AlarmDeploy.getPlcxsj();
                            plsj1 = AlarmDeploy.getPlcxsj1() == null ? -1 : AlarmDeploy.getPlcxsj1();
                            plsj2 = AlarmDeploy.getPlcxsj2() == null ? -1 : AlarmDeploy.getPlcxsj2();

                            if (plsj1 > 0) {
                                plsj1 = plsj1 * 60;
                                if (KeepTime >= plsj1) status = 1;
                            }
                            if (plsj2 > 0) {
                                plsj2 = plsj2 * 60;
                                if (KeepTime >= plsj2) status = 2;
                            }
                            if (plsj > 0) {
                                plsj = plsj * 60;
                                if (KeepTime >= plsj) status = 3;
                            }
                        }
                    }
                    break;
                    case "夜间行驶报警": {
                        int yjsj = -1;
                        int yjsj1 = -1;
                        int yjsj2 = -1;
                        if (AlarmDeploy != null) {
                            status = 1;

                            yjsj = AlarmDeploy.getYjcxsj() == null ? -1 : AlarmDeploy.getYjcxsj();
                            yjsj1 = AlarmDeploy.getYjcxsj1() == null ? -1 : AlarmDeploy.getYjcxsj1();
                            yjsj2 = AlarmDeploy.getYjcxsj2() == null ? -1 : AlarmDeploy.getYjcxsj2();

                            if (yjsj1 > 0) {
                                yjsj1 = yjsj1 * 60;
                                if (KeepTime >= yjsj1) status = 1;
                            }
                            if (yjsj2 > 0) {
                                yjsj2 = yjsj2 * 60;
                                if (KeepTime >= yjsj2) status = 2;
                            }
                            if (yjsj > 0) {
                                yjsj = yjsj * 60;
                                if (KeepTime >= yjsj) status = 3;
                            }
                        }
                    }
                    break;
                    case "无数据报警":
                    case "不定位报警": {
                        int ycsj = -1;
                        if (AlarmDeploy != null) {
                            status = 2;

                            ycsj = AlarmDeploy.getYccxsj() == null ? -1 : AlarmDeploy.getYccxsj();

                            if (ycsj > 0) {
                                if (KeepTime >= ycsj) status = 3;
                            }
                        }
                    }
                    break;
                }

                if (AlarmDeploy == null) {
                    status = 2;
                }

                if (status < VehAlarm.getAlarmLevel()) {
                    status = VehAlarm.getAlarmLevel();
                }

                BaobiaoAlarmsummary baobiaoAlarmsummary = new BaobiaoAlarmsummary();
                baobiaoAlarmsummary.setVehid(VehAlarm.getVehicleId());
                baobiaoAlarmsummary.setDeptId(VehAlarm.getDeptId());
                baobiaoAlarmsummary.setAnalyzemode(analyzeMode);
                baobiaoAlarmsummary.setAlarmtype(AlarmNote);
                baobiaoAlarmsummary.setBegintime(TimeUtils.toDate(VehAlarm.getAlarmBeginTime()));

                baobiaoAlarmsummary.setEndtime(VehAlarm.getAlarmEndTime());
                baobiaoAlarmsummary.setMaxspeed(VehAlarm.getMaxSpeed());

                if (status == 1) {
                    if (AlarmNote.equalsIgnoreCase("疲劳驾驶报警"))   //20221108华哥要求
                    {
                        baobiaoAlarmsummary.setPassed(100);
                    } else {
                        baobiaoAlarmsummary.setPassed(99);
                    }
                } else {
                    if (KeepTime < 500000) {
                        baobiaoAlarmsummary.setPassed(100);
                    } else {
                        baobiaoAlarmsummary.setPassed(-1);
                    }
                }

                baobiaoAlarmsummary.setKeeptime(KeepTime);
                baobiaoAlarmsummary.setVerifyname("System");
                if(baobiaoAlarmsummary.getVerifytime()==null) baobiaoAlarmsummary.setVerifytime(new Date());
                baobiaoAlarmsummary.setEndspeed(VehAlarm.getVelocity() != null ? VehAlarm.getVelocity() : 0);
                baobiaoAlarmsummary.setDistance(VehAlarm.getDistance() != null ? VehAlarm.getDistance() : 0.0);
                baobiaoAlarmsummary.setCutofftime(TimeUtils.toDate(TimeUtils.getNow()));
                baobiaoAlarmsummary.setEndlongitude(BigDecimal.valueOf(VehAlarm.getLongitude() != null ? VehAlarm.getLongitude() : 0.0));
                baobiaoAlarmsummary.setEndlatitude(BigDecimal.valueOf(VehAlarm.getLatitude() != null ? VehAlarm.getLatitude() : 0.0));
                baobiaoAlarmsummary.setEndroadname(VehAlarm.getRoadName());
                baobiaoAlarmsummary.setIsregionv("0");
                baobiaoAlarmsummary.setAtnocturnal(0);
                baobiaoAlarmsummary.setBaojingtype(BaoJingType);
                baobiaoAlarmsummary.setChaosubi(ChaoSuBi);
                baobiaoAlarmsummary.setIcardsign(VehAlarm.getDriver());
                baobiaoAlarmsummary.setStatus(status);
                list.add(baobiaoAlarmsummary);

                if (!AlarmNote.equalsIgnoreCase("24小时不在线报警")) {


                    BaobiaoAlarmsummaryRecord baobiaoAlarmsummaryRecord = new BaobiaoAlarmsummaryRecord();
                    baobiaoAlarmsummaryRecord.setVehid(VehAlarm.getVehicleId());
                    baobiaoAlarmsummaryRecord.setBegintime(TimeUtils.toDate(VehAlarm.getTime()));
                    baobiaoAlarmsummaryRecord.setAlarmtype(AlarmNote);
                    baobiaoAlarmsummaryRecord.setLongitude(BigDecimal.valueOf(VehAlarm.getLongitude()));
                    baobiaoAlarmsummaryRecord.setLatitude(BigDecimal.valueOf(VehAlarm.getLatitude()));
                    baobiaoAlarmsummaryRecord.setVelocity(VehAlarm.getVelocity());
                    baobiaoAlarmsummaryRecord.setAngle(VehAlarm.getAngle());
                    baobiaoAlarmsummaryRecord.setRoadLevel("0");
                    baobiaoAlarmsummaryRecord.setRoadLimited(0);
                    baobiaoAlarmsummaryRecord.setKeeptime(Long.valueOf(Math.round(KeepTime)).intValue());
                    baobiaoAlarmsummaryRecord.setLevel(status);
                    baobiaoAlarmsummaryRecord.setAlarmreportid(VehAlarm.getAlarmGuid());

                    list.add(baobiaoAlarmsummaryRecord);


                }


            }


        } catch (Exception e) {
            session.rollback();
        } finally {
            session.close();

            for (Object item : list) {
                collector.collect(item);
            }


        }


    }
}
